package com.sunyur.tailor.channel.service.middle.kafka;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * kafka 消息发送
 */
@Component
@ConditionalOnProperty(name = "spring.kafka.switch", havingValue = "open")
public class KafkaProducer {
    private static final Logger LOGGER_INFO = LoggerFactory.getLogger(KafkaProducer.class);
    private static final Logger LOGGER_ERROR = LoggerFactory.getLogger(KafkaProducer.class);

    @Autowired
    private KafkaTemplate<String, Object> oaKafkaTemplate;

    /**
     * @Description: 推送消息到OA
     */
    public <T> void send2Oa(String topic, String message) {
        try {
            LOGGER_INFO.info("KafkaProducer send2Oa topic:{}, message:{}", topic, message);
            ListenableFuture<SendResult<String, Object>> future = oaKafkaTemplate.send(topic, JSON.toJSONString(message));
            future.addCallback(
                    (result -> {
                        LOGGER_INFO.info("=====>kafka消息发送成功，结果:{}", JSON.toJSONString(result));
                    }), (result -> {
                        LOGGER_INFO.warn("=====>kafka消息发送失败，结果:{}", JSON.toJSONString(result));
                    })
            );
        } catch (Exception e) {
            LOGGER_ERROR.error("=====>kafka消息发送失败", e);
        }
    }
}
